package avicit.bdp.dcs.hdfs;

import avicit.bdp.core.constant.Constants;
import avicit.bdp.dcs.utils.PropertyUtil;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;

public class HdfsTest {

    public static void main(String[] args) throws Exception {
        System.setProperty("java.security.krb5.conf", PropertyUtil.getString("java.security.krb5.conf.path"));
        Configuration configuration = new org.apache.hadoop.conf.Configuration();
        String defaultFS = PropertyUtil.getString("fs.defaultFS");
        configuration.set("fs.defaultFS", defaultFS);
//        configuration.setBoolean("hadoop.security.authorization", true);
        configuration.set("hadoop.security.authentication", Constants.KERBEROS);
//        configuration.set("hadoop.rpc.protection", "privacy");
//        configuration.set("dfs.data.transfer.protection", "integrity");

        UserGroupInformation.setConfiguration(configuration);

        String kerberosPrincipal = PropertyUtil.getString("login.user.keytab.username");
        String kerberosKeytabFilePath = PropertyUtil.getString("login.user.keytab.path");
        try {
            UserGroupInformation.loginUserFromKeytab(kerberosPrincipal, kerberosKeytabFilePath);
            System.out.println("kerberos认证成功");
        } catch (Exception e) {
            e.printStackTrace();
            String message = String.format(
                    "kerberos认证失败,请确定kerberosKeytabFilePath[%s]和kerberosPrincipal[%s]填写正确",
                    kerberosKeytabFilePath, kerberosPrincipal);
            System.out.println(message);
        }

        try {
            FileSystem fileSystem = FileSystem.get(configuration);
            System.out.println("fileSystem = " + fileSystem);

            Path path = new Path("/griffin/persist/qq/1629958601937/_METRICS");
            System.out.println(fileSystem.exists(path));

//            fileSystem.delete(path, true);

            FSDataInputStream fsDataInputStream = fileSystem.open(path);
            byte[] array = IOUtils.toByteArray(fsDataInputStream);
            String str = new String(array);
            JSONObject jsonObject = JSONObject.parseObject(str);
            String metrics = jsonObject.getString("metrics");
            JSONObject metricsObject = JSONObject.parseObject(metrics);
            String count = metricsObject.getString("count");
            System.out.println("count = " + count);
        } catch (Exception e) {
            e.printStackTrace();
            String message = String.format("获取FileSystem失败,请检查HDFS地址是否正确: [%s]",
                    "message:defaultFS =" + defaultFS);
            System.out.println(message);
        }

    }

}
